package samples;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import util.Printer;
import util.ThreadHelper;

/**
 * @author zhengyunguang
 * 将每个module中的task取出来，通过priority排序
 * 并将排完序的task中的Observable取出来concat依次执行
 */
class Sample1 {

    public static void main(String[] args) {

        getTarget().subscribe(new Consumer<TaskResult>() {
            @Override
            public void accept(TaskResult taskResult) throws Exception {
                Printer.print(taskResult);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Printer.print(throwable);
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Printer.print("target onComplete");
            }
        });

        ThreadHelper.sleep();
    }

    public static Observable<TaskResult> getTarget() {

        List<Module> modules = new ArrayList<>();
        modules.add(new ModuleA());
        modules.add(new ModuleB());
        modules.add(new ModuleC());
        modules.add(new ModuleD());

        return Observable.fromIterable(modules)
                .filter(new Predicate<Module>() {
                    @Override
                    public boolean test(Module module) throws Exception {
                        return module.getTask() != null;
                    }
                })
                .map(new Function<Module, Task>() {
                    @Override
                    public Task apply(Module module) throws Exception {
                        return module.getTask();
                    }
                })
                .toSortedList(new Comparator<Task>() {
                    @Override
                    public int compare(Task o1, Task o2) {
                        return o2.priority - o1.priority;
                    }
                })
                .flatMapObservable(new Function<List<Task>, ObservableSource<Task>>() {
                    @Override
                    public ObservableSource<Task> apply(List<Task> tasks) throws Exception {
                        return Observable.fromIterable(tasks);
                    }
                })
                .map(new Function<Task, Observable<TaskResult>>() {
                    @Override
                    public Observable<TaskResult> apply(Task task) throws Exception {
                        return task.realTask;
                    }
                })
                .toList()
                .flatMapObservable(new Function<List<Observable<TaskResult>>, ObservableSource<TaskResult>>() {
                    @Override
                    public ObservableSource<TaskResult> apply(List<Observable<TaskResult>> observables) throws Exception {
                        return Observable.concat(observables);
                    }
                });
    }


    static class ModuleA extends Module {
        @Override
        Task getTask() {
            return new Task(1, Observable.create(new ObservableOnSubscribe<TaskResult>() {
                @Override
                public void subscribe(ObservableEmitter<TaskResult> emitter) throws Exception {
                    emitter.onNext(new TaskResult("ModuleA", true));
                    emitter.onComplete();
                }
            }));
        }
    }

    static class ModuleB extends Module {
        @Override
        Task getTask() {
            return new Task(100, Observable.create(new ObservableOnSubscribe<TaskResult>() {
                @Override
                public void subscribe(ObservableEmitter<TaskResult> emitter) throws Exception {
                    emitter.onNext(new TaskResult("ModuleB", false));
                    emitter.onComplete();
                }
            }));
        }
    }

    static class ModuleC extends Module {
        @Override
        Task getTask() {
            return null;
        }
    }

    static class ModuleD extends Module {
        @Override
        Task getTask() {
            return new Task(50, Observable.create(new ObservableOnSubscribe<TaskResult>() {
                @Override
                public void subscribe(ObservableEmitter<TaskResult> emitter) throws Exception {
                    emitter.onNext(new TaskResult("ModuleD", false));
                    emitter.onComplete();
//                    emitter.onError(new RuntimeException("测试错误"));
                }
            }).delay(10, TimeUnit.SECONDS));
        }
    }

    static class Module {
        Task getTask() {
            return null;
        }
    }

    static class Task {

        public Task(int priority, Observable<TaskResult> realTask) {
            this.priority = priority;
            this.realTask = realTask;
        }

        /**
         * 优先级
         */
        int priority;
        /**
         * 要执行的任务
         */
        Observable<TaskResult> realTask;

        @Override
        public String toString() {
            return "Task{" +
                    "priority=" + priority +
                    ", realTask=" + realTask +
                    '}';
        }
    }

    static class TaskResult {

        String name;

        public TaskResult(String name, boolean success) {
            this.name = name;
            this.success = success;
        }

        /**
         * 是否成功
         */
        boolean success;

        @Override
        public String toString() {
            return "TaskResult{" +
                    "name='" + name + '\'' +
                    ", success=" + success +
                    '}';
        }
    }
}
